package com.kitty.cross.core;

import com.kitty.common.thread.NamedThreadFactory;
import com.kitty.cross.core.callback.CallbackTask;
import com.kitty.cross.core.client.C2SSessionPoolFactory;
import com.kitty.cross.core.client.CCSession;
import com.kitty.mina.message.Message;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class CrossTransportManager {

    private int defaultCoreSum = Runtime.getRuntime().availableProcessors();

    private ExecutorService[] services;

    private ExecutorService asynService;

    @Autowired
    private C2SSessionPoolFactory sessionFactory;

    @PostConstruct
    private void init() {
        services = new ExecutorService[defaultCoreSum];
        for (int i = 0; i < defaultCoreSum; i++) {
            services[i] = Executors.newSingleThreadExecutor(new NamedThreadFactory("rpc-transport" + i));
        }

        asynService = Executors.newFixedThreadPool(defaultCoreSum);
    }

    /**
     * 同步发消息
     *
     * @param ip
     * @param port
     * @param message
     */
    public void sendMessageSync(String ip, int port, Message message) {
        CCSession session = sessionFactory.borrowSession(ip, port);
        try {
            session.sendMessage(message);
        } finally {
            sessionFactory.returnSession(session);
        }
    }

    /**
     * 异步发消息
     *
     * @param ip
     * @param port
     * @param message
     */
    public void sendMessageAsync(String ip, int port, Message message) {
        String key = (ip + port).toString();
        int index = key.hashCode() % defaultCoreSum;
        services[index].submit(() -> {
            sendMessageSync(ip, port, message);
        });
        CCSession session = sessionFactory.borrowSession(ip, port);
        sendMessageAsync(session, message);
    }

    /**
     * 异步发消息
     *
     * @param session
     * @param message
     */
    public void sendMessageAsync(CCSession session, Message message) {
        try {
            session.sendMessage(message);
        } finally {
            sessionFactory.returnSession(session);
        }
    }

    public Message callBack(CCSession session, Message message) throws ExecutionException, InterruptedException {
        CallbackTask task = CallbackTask.valueOf(session, message);
        return asynService.submit(task).get();
    }

}
